package com.daidai.cdc;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * {
 * "database":"",
 * "table":""
 * "operation":"",
 * "data":""
 * }
 */
public class CustomerSchema implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector collector) throws Exception {

        //获取 database和 table
        String topic = sourceRecord.topic();
        //分隔符得写 \\. 不然就报错
        String[] strings = topic.split("\\.");
        String database = strings[1];
        String table = strings[2];

        //获取 data
        Struct value = (Struct) sourceRecord.value();
        JSONObject data = new JSONObject();
        Struct before = value.getStruct("before");
        JSONObject beforeData = new JSONObject();
        for (Field field : before.schema().fields()) {
            Object o = before.get(field);
            beforeData.put(field.name(), o);
        }
        Struct after = value.getStruct("after");
        JSONObject afterData = new JSONObject();
        for (Field field : after.schema().fields()) {
            Object o = after.get(field);
            afterData.put(field.name(), o);
        }
        data.put("before", beforeData);
        data.put("after", afterData);

        //获取操作类型
        Envelope.Operation op = Envelope.operationFor(sourceRecord);

        //装配数据
        JSONObject object = new JSONObject();
        object.put("database", database);
        object.put("table", table);
        object.put("operation", op);
        object.put("data", data);

        collector.collect(object.toJSONString());
    }

    @Override
    public TypeInformation getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
